//The MIT License (MIT)
//
//Copyright (c) 2018 dote
//
//Permission is hereby granted, free of charge, to any person obtaining a copy of
//this software and associated documentation files (the "Software"), to deal in
//the Software without restriction, including without limitation the rights to
//use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
//the Software, and to permit persons to whom the Software is furnished to do so,
//subject to the following conditions:
//
//The above copyright notice and this permission notice shall be included in all
//copies or substantial portions of the Software.
//
//THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
//FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
//COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
//IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
//CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

package com.github.dote.mio.server;


import com.corundumstudio.socketio.AckMode;
import com.corundumstudio.socketio.Configuration;
import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.protocol.JacksonJsonSupport;
import com.github.dote.mio.ack.AckReceiveTask;
import com.github.dote.mio.ack.AckSendTask;
import com.github.dote.mio.client.IOClient;
import com.github.dote.mio.config.IOServerConfig;
import com.github.dote.mio.event.*;
import com.github.dote.mio.netty.MIOClientInboundHandler;
import com.github.dote.mio.netty.MIOServerChannel;
import com.github.dote.mio.processor.AckProcessor;
import com.github.dote.mio.processor.EventMessageProcessor;
import com.github.dote.mio.processor.KeepAliveProcessor;
import com.github.dote.mio.protocal.VoidData;
import com.github.dote.mio.server.handler.PacketHandler;
import com.github.dote.mio.session.SessionManager;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.concurrent.Future;

@Slf4j
@Getter
public class IOServer {

    private SocketIOServer socketIOServer;

    private AckSendTask ackSendTask = null;
    private AckReceiveTask ackReceiveTask=null;
    private IOServerConfig config = null;
    private PacketHandler packetHandler = null;
    private EventManager eventManager = null;
    private SessionManager sessionManager = null;
    private AckProcessor ackProcessor = null;
    private KeepAliveProcessor keepAliveProcessor = null;
    private EventMessageProcessor eventMessageProcessor = null;

    private boolean running = false;

    private final EventLoopGroup bossEventLoopGroup = new NioEventLoopGroup();
    private final EventLoopGroup workerEventLoopGroup = new DefaultEventLoopGroup();
    private Channel serverChannel = null;

    public IOServer(IOServerConfig config) {
        this.config = config;
        init();
    }

    public boolean isRunning() {
        return running;
    }

    private void init(){
        packetHandler = new PacketHandler(this);
        ackSendTask = new AckSendTask( this);
        ackReceiveTask = new AckReceiveTask( this);
        eventManager = new EventManager();
        sessionManager = new SessionManager(this);
        keepAliveProcessor = new KeepAliveProcessor();
        ackProcessor = new AckProcessor(ackSendTask,sessionManager);
        eventMessageProcessor = new EventMessageProcessor(sessionManager, eventManager,ackReceiveTask);

        ackSendTask.startup(true);
        ackReceiveTask.startup(false);

        initEvent();
    }

    private void initEvent() {
        this.eventManager.addEventListener("exit", new EventListener<VoidData>() {

            @Override
            public void onEvent(IOClient client, VoidData data) {
                IOServer.this.packetHandler.disconnect(client);
            }
        });
    }

    /**
     * startup
     *
     * @throws Exception
     */
    public void startup() throws Exception {

        if (!isRunning()) {

            ServerBootstrap bootstrap = initServerBootstrap();

            ChannelFuture cf = bootstrap.bind("0.0.0.0", config.getUdpPort()).syncUninterruptibly();
            serverChannel = cf.channel();

            this.running = true;

            log.info("Successfully start m-io server,port:" + config.getUdpPort());

            //打开socketio 服务端
            if(config.isEnableSocketIOServer()){
                SocketConfig socketConfig=new SocketConfig();
                socketConfig.setTcpNoDelay(config.isTcpNoDelay());
                socketConfig.setAcceptBackLog(config.getAcceptBackLog());
                socketConfig.setReuseAddress(config.isReuseAddress());
                socketConfig.setSoLinger(config.getSoLinger());
                socketConfig.setTcpKeepAlive(config.isTcpKeepAlive());
                socketConfig.setTcpReceiveBufferSize(config.getTcpReceiveBufferSize());
                socketConfig.setTcpSendBufferSize(config.getTcpSendBufferSize());

                Configuration socketIOConfiguration=new Configuration();
                socketIOConfiguration.setSocketConfig(socketConfig);
                socketIOConfiguration.setHostname("0.0.0.0");
                socketIOConfiguration.setPort(config.getSocketIOPort());

                socketIOConfiguration.setAckMode(AckMode.AUTO);



                socketIOServer=new SocketIOServer(socketIOConfiguration);
                Future<Void> f= socketIOServer.startAsync();
                ((io.netty.util.concurrent.Future<Void>) f).addListener(new GenericFutureListener<io.netty.util.concurrent.Future<? super Void>>() {
                    @Override
                    public void operationComplete(io.netty.util.concurrent.Future<? super Void> future) throws Exception {
                                if(future.isSuccess()){
                                    log.info("Successfully start socketio server,port:" + config.getSocketIOPort());
                                    packetHandler.socketIOEventInit();
                                }else{
                                    log.info("Failure start socketio server,port:" + config.getSocketIOPort());
                                }
                    }
                });
            }

            serverChannel.closeFuture().await();
        } else {
            if (config.isDebug())
                log.debug("The m-io server has started,Please shutdown and retry");
        }
    }

    /**
     * shutdown
     */
    public void shutdown() {
        if (serverChannel != null)
            serverChannel.close();

        bossEventLoopGroup.shutdownGracefully();
        workerEventLoopGroup.shutdownGracefully();

        ackSendTask.stop();
        ackReceiveTask.stop();

        this.running = false;
    }

    /**
     * 初始化netty serve
     *
     * @return
     */
    private ServerBootstrap initServerBootstrap() {
        return new ServerBootstrap()
                .group(bossEventLoopGroup, workerEventLoopGroup)
                .channel(MIOServerChannel.class)
                .childHandler(initChildChannelHandler4Netty());
    }

    protected ChannelHandler initChildChannelHandler4Netty() {
        return new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                channel.pipeline()
                        .addLast(new ReadTimeoutHandler(config.getReadTimeout()))
                        .addLast(new MIOClientInboundHandler(packetHandler));
            }
        };
    }

    public void addConnectionListener(ConnectionListener listener){
        getEventManager().addConnectionListener(listener);
    }

    public void addDisconnectionListener(DisconnectionListener listener){
        getEventManager().addDisconnectionListener(listener);
    }

    public void addLostPacketListener(LostPacketListener listener){
        getEventManager().addLostPacketListener(listener);
    }

    public void addOfflinePakcetListener(OfflinePacketListener listener){
        getEventManager().addOfflinePakcetListener(listener);
    }

    public void addEventListener(String eventName, EventListener<?> listener){
        getEventManager().addEventListener(eventName,listener);
    }

    public void removeEventListener(String eventName){
        getEventManager().removeEventListener(eventName);
    }

    public Collection<IOClient> getAllClients(){
        return sessionManager.getAllClients();
    }
}
